package com.pisolution.scheduler.runtime.registry;

import com.pisolution.scheduler.core.utils.JsonUtils;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Nullable;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 带有缓存措施的注册中心交互客户端.
 *
 * @author sora
 * @version 0.1.0
 */
@Component
public class CachedZookeeperClient extends ZookeeperClient {
    private static final Logger LOGGER = LoggerFactory.getLogger(CachedZookeeperClient.class);

    // -------------------------------------------------------------------------------------------------

    /** 缓存交互介质. */
    private CuratorCache cache;

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    public CachedZookeeperClient(final ZookeeperConfig config) {
        super(config);
    }

    // -------------------------------------------------------------------------------------------------
    //  生命周期
    // -------------------------------------------------------------------------------------------------

    @Override
    public void close() {
        // 1. 清除本地缓存.
        this.cache.close();

        // 2. 关闭连接.
        super.close();
    }

    // -------------------------------------------------------------------------------------------------
    //  辅助方法 (生命周期)
    // -------------------------------------------------------------------------------------------------

    @Override
    protected void nodeCache() {
        // 1. 初始化缓存实例 监控管理节点与工作节点注册路径.
        this.cache = CuratorCache.build(super.getClient(), "/servers");

        // 2. 同步缓存.
        this.cache.start();
    }

    @Override
    protected void registerListener() {
        this.cache.listenable().addListener((type, oldData, data) -> {
            // 1. 如果无法获取发生事件的节点路径则忽略该事件.
            if (data.getPath() == null) {
                return;
            }

            // 2. 传递节点事件.
            this.changed(this.getClient(), type, data.getPath());
        });
    }

    /**
     * 处理节点事件.
     *
     * @param client 用于与高可用服务进行交互的客户端.
     * @param type 需要处理的事件的类型.
     * @param path 发生事件的节点路径.
     */
    protected void changed(final CuratorFramework client, final CuratorCacheListener.Type type, final String path) {
    }

    // -------------------------------------------------------------------------------------------------
    //  辅助方法 (数据交互)
    // -------------------------------------------------------------------------------------------------

    /**
     * 获取给定类型的服务节点在高可用服务中的存储路径.
     *
     * @param type 服务节点类型.
     *
     * @return 该类型的节点在高可用服务中的存储路径.
     */
    protected String getNodePath(final NodeType type) {
        switch (type) {
            case MASTER:
                return "/servers/master";
            case WORKER:
                return "/servers/worker";
            case DEAD_SERVER:
                return "/dead-servers";
            default:
                return "";
        }
    }

    /**
     * 获取所有符合给定类型的服务节点存储路径及其存储的数据.
     *
     * @param type 待查询的节点类型信息.
     *
     * @return 以集合形式存储的所有符合该类型的节点信息.
     */
    protected Map<String, String> getServerMap(final NodeType type) {
        // 1. 创建一个临时容器用于存储相关节点信息.
        final Map<String, String> servers = new HashMap<>();

        try {
            // 2. 尝试获取所有符合该类型的节点的存储路径.
            final String path = this.getNodePath(type);
            final List<String> pathList = super.children(path);

            // 3. 查询每一个符合类型的节点中所存储的数据.
            for (final String server: pathList) {
                servers.putIfAbsent(server, super.get(path + "/" + server));
            }
        } catch (Exception e) {
            LOGGER.error("Could not iterate the server list.", e);
        }

        return servers;
    }

    /**
     * 获取注册中心中所有给定类型的节点信息.
     *
     * @param type 查询的节点所属类型.
     *
     * @return 符合该类型的所有节点信息.
     */
    @SuppressWarnings("SameParameterValue")
    protected List<Server> getServerList(NodeType type) {
        // 1. 获取 {host: data} 格式的符合给定类型的节点信息以及该类型节点所在的父级目录.
        final Map<String, String> servers = this.getServerMap(type);
        final String parentPath = this.getNodePath(type);

        // 2. 创建一个临时集合用于存储符合要求的节点信息.
        final List<Server> serverList = new ArrayList<>();

        for (Map.Entry<String, String> entry: servers.entrySet()) {
            // 3. 从心跳数据包中解析出节点信息.
            final Server server = this.parseServerFromHeartbeat(entry.getValue());

            // 4. 校验解析出的节点信息是否为空.
            if (server == null) {
                continue;
            }

            // 5. 填充节点信息.
            final String[] quorums = entry.getKey().split(":");
            final String[] hosts = quorums[0].split("/");

            server.setDirectory(parentPath + "/" + entry.getKey());
            server.setHost(hosts[hosts.length - 1]);
            server.setPort(Integer.parseInt(quorums[1]));

            serverList.add(server);
        }

        return serverList;
    }

    /**
     * 从给定的心跳数据包中解析出包含的节点信息.
     *
     * @param heartbeat 给定的心跳数据包.
     *
     * @return 解析出来的节点信息.
     */
    @Nullable
    private Server parseServerFromHeartbeat(String heartbeat) {
        // 1. 校验心跳数据包是否为空.
        if (StringUtils.isBlank(heartbeat)) {
            return null;
        }

        // 2. 校验心跳数据包是否不完整 (心跳数据包共10段内容).
        final String[] masters = heartbeat.split(",");
        if (masters.length != 10) {
            return null;
        }

        // 3. 尝试解析心跳数据包.
        final Server server = new Server();

        try {
            // 4. 解析系统资源信息.
            final Map<String, String> resources = new HashMap<>();

            resources.put("cpuUsage", masters[0]);
            resources.put("memoryUsage", masters[1]);
            resources.put("loadAverage", masters[2]);

            // 5. 解析其他信息.
            server.setPid(Integer.parseInt(masters[9]));
            server.setResourceInfo(JsonUtils.serializeToString(resources));
            server.setCreateTime(DateUtils.parseDate(masters[6], "yyyy-MM-dd HH:mm:ss"));
            server.setLastHeartbeatTime(DateUtils.parseDate(masters[7], "yyyy-MM-dd HH:mm:ss"));
        } catch (ParseException ignore) {
            // 该异常无需处理.
        }

        return server;
    }
}
